package net.opentsdb.tsd;

import org.jboss.netty.channel.*;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Hujf
 * @title: ConnectionManager
 * @date 2021-09-15 11:28
 * @description: 连接管理器   处理各种连接的handler
 */
public class ConnectionManager extends SimpleChannelHandler {

    private static final Logger LOG = LoggerFactory.getLogger(ConnectionManager.class);

    private static final AtomicLong connections_established = new AtomicLong();

    /**
     * Max connections can be serviced by tsd, if over limit, tsd will refuse
     * new connections.
     */
    private final int connections_limit;

    private final AtomicInteger open_connections = new AtomicInteger();

    private static final DefaultChannelGroup channels = new DefaultChannelGroup("all-channels");

    public ConnectionManager(int connections_limit) {
        LOG.info("TSD concurrent connection limit set to: " + connections_limit);
        this.connections_limit = connections_limit;
    }

    @Override
    public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        if (connections_limit > 0) {
            final int channel_size = open_connections.incrementAndGet();
            if (channel_size > connections_limit) {
                throw new ConnectionRefusedException("Channel size (" + channel_size + ") exceeds total "
                        + "connection limit (" + connections_limit + ")");
            }
        }
        channels.add(e.getChannel());
        connections_established.incrementAndGet();
    }

    /**
     * Simple exception for refusing a connection.
     */
    private static class ConnectionRefusedException extends ChannelException {

        /**
         * Default ctor with a message.
         *
         * @param message A descriptive message for the exception.
         */
        public ConnectionRefusedException(final String message) {
            super(message);
        }

        private static final long serialVersionUID = 5348377149312597939L;
    }

    @Override
    public void handleUpstream(final ChannelHandlerContext ctx,
                               final ChannelEvent e) throws Exception {
        if (e instanceof ChannelStateEvent) {
            LOG.info(e.toString());
        }
        super.handleUpstream(ctx, e);
    }

}
